package org.apache.lucene.search;

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import java.io.Closeable;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexReader; // javadocs
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.IndexSearcher; // javadocs
import org.apache.lucene.store.Directory;
import org.apache.lucene.util.IOUtils;
import org.apache.lucene.util.ThreadInterruptedException;

/**
 * Utility class to manage sharing near-real-time searchers
 * across multiple searching threads.
 *
 * <p>NOTE: to use this class, you must call {@link #maybeReopen(boolean)}
 * periodically.  The {@link NRTManagerReopenThread} is a
 * simple class to do this on a periodic basis.  If you
 * implement your own reopener, be sure to call {@link
 * #addWaitingListener} so your reopener is notified when a
 * caller is waiting for a specific generation searcher. </p>
 *
 * @lucene.experimental
 */

public class NRTManager implements Closeable {
    private static final long MAX_SEARCHER_GEN = Long.MAX_VALUE;
    private final IndexWriter writer;
    private final SearcherManagerRef withoutDeletes;
    private final SearcherManagerRef withDeletes;
    private final AtomicLong indexingGen;
    private final List<WaitingListener> waitingListeners = new CopyOnWriteArrayList<WaitingListener>();
    private final ReentrantLock reopenLock = new ReentrantLock();
    private final Condition newGeneration = reopenLock.newCondition();

    /**
     * Create new NRTManager.
     * 
     *  @param writer IndexWriter to open near-real-time
     *         readers
     *  @param warmer optional {@link SearcherWarmer}.  Pass
     *         null if you don't require the searcher to warmed
     *         before going live.  If this is non-null then a
     *         merged segment warmer is installed on the
     *         provided IndexWriter's config.
     *
     *  <p><b>NOTE</b>: the provided {@link SearcherWarmer} is
     *  not invoked for the initial searcher; you should
     *  warm it yourself if necessary.
     */
    public NRTManager(IndexWriter writer, SearcherWarmer warmer) throws IOException {
        this(writer, null, warmer, true);
    }

    /**
     * Create new NRTManager.
     * 
     *  @param writer IndexWriter to open near-real-time
     *         readers
     *  @param es optional ExecutorService so different segments can
     *         be searched concurrently (see {@link IndexSearcher#IndexSearcher(IndexReader, ExecutorService)}.
     *         Pass <code>null</code> to search segments sequentially.
     *  @param warmer optional {@link SearcherWarmer}.  Pass
     *         null if you don't require the searcher to warmed
     *         before going live.  If this is non-null then a
     *         merged segment warmer is installed on the
     *         provided IndexWriter's config.
     *
     *  <p><b>NOTE</b>: the provided {@link SearcherWarmer} is
     *  not invoked for the initial searcher; you should
     *  warm it yourself if necessary.
     */
    public NRTManager(IndexWriter writer, ExecutorService es, SearcherWarmer warmer) throws IOException {
        this(writer, es, warmer, true);
    }

    /**
     * Expert: just like {@link
     * #NRTManager(IndexWriter,ExecutorService,SearcherWarmer)},
     * but you can also specify whether every searcher must
     * apply deletes.  This is useful for cases where certain
     * uses can tolerate seeing some deleted docs, since
     * reopen time is faster if deletes need not be applied. */
    public NRTManager(IndexWriter writer, ExecutorService es, SearcherWarmer warmer, boolean alwaysApplyDeletes) throws IOException {
        this.writer = writer;
        if (alwaysApplyDeletes) {
            withoutDeletes = withDeletes = new SearcherManagerRef(true, 0, new SearcherManager(writer, true, warmer, es));
        } else {
            withDeletes = new SearcherManagerRef(true, 0, new SearcherManager(writer, true, warmer, es));
            withoutDeletes = new SearcherManagerRef(false, 0, new SearcherManager(writer, false, warmer, es));
        }
        indexingGen = new AtomicLong(1);
    }

    /** NRTManager invokes this interface to notify it when a
     *  caller is waiting for a specific generation searcher
     *  to be visible. */
    public static interface WaitingListener {
        public void waiting(boolean requiresDeletes, long targetGen);
    }

    /** Adds a listener, to be notified when a caller is
     *  waiting for a specific generation searcher to be
     *  visible. */
    public void addWaitingListener(WaitingListener l) {
        waitingListeners.add(l);
    }

    /** Remove a listener added with {@link
     *  #addWaitingListener}. */
    public void removeWaitingListener(WaitingListener l) {
        waitingListeners.remove(l);
    }

    public long updateDocument(Term t, Document d, Analyzer a) throws IOException {
        writer.updateDocument(t, d, a);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long updateDocument(Term t, Document d) throws IOException {
        writer.updateDocument(t, d);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long updateDocuments(Term t, Collection<Document> docs, Analyzer a) throws IOException {
        writer.updateDocuments(t, docs, a);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long updateDocuments(Term t, Collection<Document> docs) throws IOException {
        writer.updateDocuments(t, docs);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long deleteDocuments(Term t) throws IOException {
        writer.deleteDocuments(t);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long deleteDocuments(Term... terms) throws IOException {
        writer.deleteDocuments(terms);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long deleteDocuments(Query q) throws IOException {
        writer.deleteDocuments(q);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long deleteDocuments(Query... queries) throws IOException {
        writer.deleteDocuments(queries);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long deleteAll() throws IOException {
        writer.deleteAll();
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long addDocument(Document d, Analyzer a) throws IOException {
        writer.addDocument(d, a);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long addDocuments(Collection<Document> docs, Analyzer a) throws IOException {
        writer.addDocuments(docs, a);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long addDocument(Document d) throws IOException {
        writer.addDocument(d);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long addDocuments(Collection<Document> docs) throws IOException {
        writer.addDocuments(docs);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long addIndexes(Directory... dirs) throws CorruptIndexException, IOException {
        writer.addIndexes(dirs);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    public long addIndexes(IndexReader... readers) throws CorruptIndexException, IOException {
        writer.addIndexes(readers);
        // Return gen as of when indexing finished:
        return indexingGen.get();
    }

    /**
     * Waits for a given {@link SearcherManager} target generation to be available
     * via {@link #getSearcherManager(boolean)}. If the current generation is less
     * than the given target generation this method will block until the
     * correspondent {@link SearcherManager} is reopened by another thread via
     * {@link #maybeReopen(boolean)} or until the {@link NRTManager} is closed.
     * 
     * @param targetGen the generation to wait for
     * @param requireDeletes <code>true</code> iff the generation requires deletes to be applied otherwise <code>false</code>
     * @return the {@link SearcherManager} with the given target generation
     */
    public SearcherManager waitForGeneration(long targetGen, boolean requireDeletes) {
        return waitForGeneration(targetGen, requireDeletes, -1, TimeUnit.NANOSECONDS);
    }

    /**
     * Waits for a given {@link SearcherManager} target generation to be available
     * via {@link #getSearcherManager(boolean)}. If the current generation is less
     * than the given target generation this method will block until the
     * correspondent {@link SearcherManager} is reopened by another thread via
     * {@link #maybeReopen(boolean)}, the given waiting time has elapsed, or until
     * the {@link NRTManager} is closed.
     * <p>
     * NOTE: if the waiting time elapses before the requested target generation is
     * available the latest {@link SearcherManager} is returned instead.
     * 
     * @param targetGen
     *          the generation to wait for
     * @param requireDeletes
     *          <code>true</code> iff the generation requires deletes to be
     *          applied otherwise <code>false</code>
     * @param time
     *          the time to wait for the target generation
     * @param unit
     *          the waiting time's time unit
     * @return the {@link SearcherManager} with the given target generation or the
     *         latest {@link SearcherManager} if the waiting time elapsed before
     *         the requested generation is available.
     */
    public SearcherManager waitForGeneration(long targetGen, boolean requireDeletes, long time, TimeUnit unit) {
        try {
            final long curGen = indexingGen.get();
            if (targetGen > curGen) {
                throw new IllegalArgumentException("targetGen=" + targetGen + " was never returned by this NRTManager instance (current gen=" + curGen + ")");
            }
            reopenLock.lockInterruptibly();
            try {
                if (targetGen > getCurrentSearchingGen(requireDeletes)) {
                    for (WaitingListener listener : waitingListeners) {
                        listener.waiting(requireDeletes, targetGen);
                    }
                    while (targetGen > getCurrentSearchingGen(requireDeletes)) {
                        if (!waitOnGenCondition(time, unit)) {
                            return getSearcherManager(requireDeletes);
                        }
                    }
                }

            } finally {
                reopenLock.unlock();
            }
        } catch (InterruptedException ie) {
            throw new ThreadInterruptedException(ie);
        }
        return getSearcherManager(requireDeletes);
    }

    private boolean waitOnGenCondition(long time, TimeUnit unit) throws InterruptedException {
        assert reopenLock.isHeldByCurrentThread();
        if (time < 0) {
            newGeneration.await();
            return true;
        } else {
            return newGeneration.await(time, unit);
        }
    }

    /** Returns generation of current searcher. */
    public long getCurrentSearchingGen(boolean applyAllDeletes) {
        if (applyAllDeletes) {
            return withDeletes.generation;
        } else {
            return Math.max(withoutDeletes.generation, withDeletes.generation);
        }
    }

    public boolean maybeReopen(boolean applyAllDeletes) throws IOException {
        if (reopenLock.tryLock()) {
            try {
                final SearcherManagerRef reference = applyAllDeletes ? withDeletes : withoutDeletes;
                // Mark gen as of when reopen started:
                final long newSearcherGen = indexingGen.getAndIncrement();
                boolean setSearchGen = false;
                if (reference.generation == MAX_SEARCHER_GEN) {
                    newGeneration.signalAll(); // wake up threads if we have a new generation
                    return false;
                }
                if (!(setSearchGen = reference.manager.isSearcherCurrent())) {
                    setSearchGen = reference.manager.maybeReopen();
                }
                if (setSearchGen) {
                    reference.generation = newSearcherGen;// update searcher gen
                    newGeneration.signalAll(); // wake up threads if we have a new generation
                }
                return setSearchGen;
            } finally {
                reopenLock.unlock();
            }
        }
        return false;
    }

    /**
     * Close this NRTManager to future searching. Any searches still in process in
     * other threads won't be affected, and they should still call
     * {@link SearcherManager#release(IndexSearcher)} after they are done.
     * 
     * <p>
     * <b>NOTE</b>: caller must separately close the writer.
     */
    public void close() throws IOException {
        reopenLock.lock();
        try {
            try {
                IOUtils.close(withDeletes, withoutDeletes);
            } finally { // make sure we signal even if close throws an exception
                newGeneration.signalAll();
            }
        } finally {
            reopenLock.unlock();
            assert withDeletes.generation == MAX_SEARCHER_GEN && withoutDeletes.generation == MAX_SEARCHER_GEN;
        }
    }

    /**
     * Returns a {@link SearcherManager}. If <code>applyAllDeletes</code> is
     * <code>true</code> the returned manager is guaranteed to have all deletes
     * applied on the last reopen. Otherwise the latest manager with or without deletes
     * is returned.
     */
    public SearcherManager getSearcherManager(boolean applyAllDeletes) {
        if (applyAllDeletes) {
            return withDeletes.manager;
        } else {
            if (withDeletes.generation > withoutDeletes.generation) {
                return withDeletes.manager;
            } else {
                return withoutDeletes.manager;
            }
        }
    }

    static final class SearcherManagerRef implements Closeable {
        final boolean applyDeletes;
        volatile long generation;
        final SearcherManager manager;

        SearcherManagerRef(boolean applyDeletes, long generation, SearcherManager manager) {
            super();
            this.applyDeletes = applyDeletes;
            this.generation = generation;
            this.manager = manager;
        }

        public void close() throws IOException {
            generation = MAX_SEARCHER_GEN; // max it out to make sure nobody can wait on another gen
            manager.close();
        }
    }
}
